package org.jpush.core;

import java.util.LinkedList;
import java.util.List;

/**
 * A <code>ChannelStreamerManager</code> will maintain
 * a list of <code>ChannelStreamer</code>, when a connection
 * arrives, it will be assigned to a proper ChannelStreamer
 * to process
 * 
 * @author Wang Lei
 *
 */
public class ChannelStreamerManager
{
	protected ChannelManager channelManager;
	private int nStreamers;
	
	public ChannelStreamerManager(ChannelManager broadcastManager)
	{
		this.nStreamers = 1;
		this.channelManager = broadcastManager;
	}

	private List<ChannelStreamer> streamersList = new LinkedList<ChannelStreamer>();


	/**
	 * Create a list of <code>BroadcastStreamer</code>
	 * and start each of them
	 */
	public void start()
	{
	    for (int i = 0; i < nStreamers; i++)
	    {
	        ChannelStreamer streamer = new ChannelStreamer(channelManager);
	        streamersList.add(streamer);
	        streamer.start();
	    }
	}

	/**
	 * Stop all the <code>BroadcastStreamer</code>
	 */
	public void stop()
	{
		ChannelStreamer streamer = null;
		while(!streamersList.isEmpty())
		{
			streamer = streamersList.get(0);
			streamer.stop();
			streamersList.remove(0);
		}
	}

	/**
	 * Assign the connection to a proper <code>BroadcastStreamer</code>
	 * 
	 * @param logicalConnection The connection will be assigned
	 */
	public void addConnection(LogicalConnection logicalConnection)
	{		
		int iListSize = streamersList.size();
		if (iListSize > 0)
		{
			ChannelStreamer streamer = null;
			ChannelStreamer lessBusy = streamersList.get(0);
			for(int idx = 1; idx < iListSize; idx ++)
			{
				streamer = streamersList.get(idx);
				if (lessBusy.getItemsCount() > streamer.getItemsCount())
				{
					lessBusy = streamer;
				}
			}
			
			lessBusy.addItem(logicalConnection);
		}
	}

	/**
	 * Reshuffle all the ChannelStreamers
	 */
	public void reshuffle()
	{
		ChannelStreamer streamer = null;
		for(int idx = 0; idx < streamersList.size(); idx ++)
		{
			streamer = streamersList.get(idx);
			streamer.reshuffle();
		}
	}
	
	/**
	 * Set how many ChannelStreamer there are.
	 * Note: Only called before start() will take effect
	 * 
	 * @param nStreamers the count of BroadcastStreamer
	 */
	public void SetStreamersCount(int nStreamers)
	{
		this.nStreamers = nStreamers;
	}
}
